package com.cimu.elk.custom;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.UnsynchronizedAppenderBase;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

/**
 * Title: KafkaAppender
 * Copyright: Copyright (c) 2017
 * <p>
 * date 2019年01月09日 12:22
 */
public class KafkaAppender extends UnsynchronizedAppenderBase<ILoggingEvent> {
    private String topic;
    private String brokerServer;
    //kafka生产者
    private Producer<String, String> producer;

    @Override
    public void start() {
        super.start();
        Properties props = new Properties();
        props.put("bootstrap.servers", this.brokerServer);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(props);
    }

    @Override
    public void stop() {
        super.stop();
        producer.close();
    }

    @Override
    protected void append(ILoggingEvent event) {
        String msg = event.getMessage();
        //拼接消息内容
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(
                this.topic, 0, "key", msg);
        System.out.println("[推送数据]:" + producerRecord);
        //发送kafka的消息
        producer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                //监听发送结果
                if (exception != null) {
                    exception.printStackTrace();
                } else {
                    System.out.println("[推送数据到kafka成功]:" + metadata);
                }
            }
        });
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getBrokerServer() {
        return brokerServer;
    }

    public void setBrokerServer(String brokerServer) {
        this.brokerServer = brokerServer;
    }

    public Producer<String, String> getProducer() {
        return producer;
    }

    public void setProducer(Producer<String, String> producer) {
        this.producer = producer;
    }
}
